package com.iot.app.flink.util;

import com.iot.app.flink.vo.IoTData;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.shaded.guava18.com.google.common.collect.Iterables;
import org.apache.flink.util.Collector;

public class UniqueVehicleState extends RichFlatMapFunction <IoTData, Tuple2<IoTData, Boolean>> {

    /**
     * The ValueState handle. The first field is the count, the second field a running sum.
     */
    private transient ListState<String> uniqueVehicleId;

    @Override
    public void flatMap(IoTData ioTData, Collector<Tuple2<IoTData, Boolean>> out) throws Exception {

        if(!Iterables.contains(uniqueVehicleId.get(),ioTData.getVehicleId())) {
            uniqueVehicleId.add(ioTData.getVehicleId());
            out.collect(Tuple2.of(ioTData, true));
            return;
        }

        out.collect(Tuple2.of(ioTData, false));
        return;


    }

    @Override
    public void open(Configuration parameters) throws Exception {

        StateTtlConfig stateTtlConfig = StateTtlConfig.newBuilder(Time.seconds(3600))
                .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
                .setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired)
                .cleanupFullSnapshot()
                .build();

        ListStateDescriptor<String> descriptor = new ListStateDescriptor<String>("unique_vehicle", String.class);
        descriptor.enableTimeToLive(stateTtlConfig);

        uniqueVehicleId = getRuntimeContext().getListState(descriptor);
    }
}
